package spring.cloud.tasks.executor.task;

import com.google.common.base.Strings;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.data.Stat;
import org.springframework.util.PropertyPlaceholderHelper;
import spring.cloud.tasks.executor.api.BizTask;
import spring.cloud.tasks.executor.api.ReturnCode;
import spring.cloud.tasks.executor.api.TaskExecuteResultGroup;
import spring.cloud.tasks.executor.exception.JobException;
import spring.cloud.tasks.executor.executor.TaskInitService;
import spring.cloud.tasks.executor.listener.config.TaskConfigurationService;
import spring.cloud.tasks.executor.listener.failover.FailoverService;
import spring.cloud.tasks.executor.listener.report.ReportService;
import spring.cloud.tasks.executor.listener.sharding.ShardingService;
import spring.cloud.tasks.executor.listener.task_execute.TaskExecutorInfoService;
import spring.cloud.tasks.executor.node.TaskNodePaths;
import spring.cloud.tasks.executor.task.statistics.ProcessCountStatistics;
import spring.cloud.tasks.executor.trigger.Scheduler;
import spring.cloud.tasks.executor.trigger.Trigger;
import spring.cloud.tasks.executor.trigger.Triggered;
import spring.cloud.tasks.executor.utils.log.LogUtils;

import java.util.*;
import java.util.concurrent.Future;

@Slf4j
@Setter
@Getter
public class TaskExecuteEngine {
    //
    protected String nameSpace;
    protected String executorId;
    protected String taskId;
    //
    protected TaskConfigurationService taskConfigurationService;
    protected TaskExecuteContextService executeContextService;
    protected TaskInitService taskInitService;
    protected TaskExecutorInfoService taskExecutorInfoService;
    protected ExecuteService executeService;
    protected FailoverService failoverService;
    protected ShardingService shardingService;
    //
    protected Scheduler scheduler;
    protected TaskExecutor taskExecutor;

    protected ReportService reportService;

    private volatile boolean stopped = false;
    private volatile boolean forceStopped = false;
    private volatile boolean aborted = false;
    private volatile boolean running = false;
    private Map<Integer, TaskItemExecuteTask> taskItemIndexToTaskItemExecuteTaskMap = new HashMap<>();
    private Object jobInstance = null;

    /**
     * 重置作业调用一次的生命周期内的变量
     */
    private void reset() {
        stopped = false;
        forceStopped = false;
        aborted = false;
        running = true;
    }


    public void shutdown() {
        if (scheduler != null) {
            // 关闭调度器
            scheduler.shutdown();
            // 关闭执行业务的线程池，使得不能再提交新的业务任务
            taskExecutor.shutdownExecutorService();
            // 检查调度器任务是否完成，如果没有，中止业务
            if (!scheduler.isTerminated()) {
                {
                    abort();
                    scheduler.awaitTermination(500L);
                }
            }
        }
    }


    protected void init() {
        Class<? extends Trigger> triggerClass = taskConfigurationService.getTaskType().getTriggerClass();
        Trigger trigger = null;
        try {
            trigger = triggerClass.newInstance();
            trigger.init(this);
        } catch (Exception e) {
            LogUtils.error(log, taskId, "Trigger init failed", e);
            throw new JobException(e);
        }
        scheduler = new Scheduler(this, trigger);
        scheduler.start();
        createJobInstanceIfNecessary();


    }

    private void createJobInstanceIfNecessary() {
        String taskClass = taskConfigurationService.getTaskConfiguration().getTaskClass();
        //jobInstance
    }


    public final void execute(final Triggered triggered) {
        LogUtils.debug(log, taskId, "Saturn start to execute job [{}]", taskId);
        // 对每一个jobScheduler，作业对象只有一份，多次使用，所以每次开始执行前先要reset
        reset();

        if (taskConfigurationService == null) {
            LogUtils.warn(log, taskId, "configService is null");
            return;
        }

        TaskExecuteContext taskExecuteContext = null;
        try {
            if (!taskConfigurationService.isEnabled() || failoverService.getFailoverTaskItemIndexList().isEmpty()) {
                shardingService.shardingIfNecessary();
            }

            if (!taskConfigurationService.isEnabled()) {
                LogUtils.debug(log, taskId, "{} is disabled, cannot be continued, do nothing about business.",
                        taskId);
                return;
            }

            taskExecuteContext = executeContextService.getTaskExecuteContext(triggered);
            if (taskExecuteContext.getTaskItemIndexList() == null || taskExecuteContext.getTaskItemIndexList().isEmpty()) {
                LogUtils.debug(log, taskId, "{} 's items of the executor is empty, do nothing about business.", taskId);
                return;
            }

            if (taskConfigurationService.isInPausePeriod()) {
                LogUtils.info(log, taskId, "the job {} current running time is in pausePeriod, do nothing about business.", taskId);
                return;
            }

            executeInternal(taskExecuteContext);

            if (isFailoverSupported() && taskConfigurationService.isFailover() && !stopped && !forceStopped && !aborted) {
                failoverService.failoverIfNecessary();
            }

            LogUtils.debug(log, taskId, "Saturn finish to execute job [{}], sharding context:{}.", taskId, taskExecuteContext);
        } catch (Exception e) {
            LogUtils.warn(log, taskId, e.getMessage(), e);
        } finally {
            running = false;
        }
    }

    private void executeInternal(final TaskExecuteContext taskExecuteContext) throws Exception {

        executeService.registerTaskItemBegin(taskExecuteContext);

        try {
            execute(taskExecuteContext);
        } finally {
            List<Integer> taskItemIndexList = taskExecuteContext.getTaskItemIndexList();
            if (!taskItemIndexList.isEmpty()) {
                Date nextFireTimePausePeriodEffected = taskExecutor.getNextFireTimePausePeriodEffected();
                boolean isEnabledReport = taskConfigurationService.isEnabled();
                for (int taskItemIndex : taskItemIndexList) {
                    if (isEnabledReport && !checkIfZooKeeperLostAfterExecute(taskItemIndex)) {
                        continue;
                    }
                    if (!aborted) {
                        executeService.registerTaskItemCompleted(taskExecuteContext, taskItemIndex, nextFireTimePausePeriodEffected);
                        if (isFailoverSupported() && taskConfigurationService.isFailover()) {
                            failoverService.updateFailoverComplete(taskItemIndex);
                        }
                    }
                }
            }

        }
    }

    protected final void execute(final TaskExecuteContext taskExecuteContext) {

        long start = System.currentTimeMillis();

        //
        // items为需要处理的作业分片
        List<Integer> taskItemIndexList = taskExecuteContext.getTaskItemIndexList();
        LogUtils.info(log, taskId, "Job {} handle items: {}", taskId, taskItemIndexList);
        // shardingItemParameters为参数表解析出来的Key/Value值
        Map<Integer, String> taskItemToTaskItemAndParameterMap = taskExecuteContext.getTaskItemIndexToTaskItemAndParameterMap();
        LogUtils.info(log, taskId, "Job {} handle items: {}", taskId, taskItemToTaskItemAndParameterMap);
        //
        Map<Integer, TaskItemExecuteResult> taskItemIndexToTaskExecuteResultMap = new HashMap<Integer, TaskItemExecuteResult>();
        for (Integer taskItemIndex : taskItemIndexList) {
            // 兼容配置错误，如配置3个分片, 参数表配置为0=*, 2=*, 则1分片不会执行
            if (!taskItemToTaskItemAndParameterMap.containsKey(taskItemIndex)) {
                LogUtils.error(log, taskId, "The {} item's parameter is not valid, will not execute the business code, please check shardingItemParameters", taskItemIndexList);
                TaskItemExecuteResult taskExecuteResult = new TaskItemExecuteResult(ReturnCode.SYSTEM_FAIL, "Config of parameter is not valid, check shardingItemParameters", TaskExecuteResultGroup.FAIL);
                taskItemIndexToTaskExecuteResultMap.put(taskItemIndex, taskExecuteResult);
            }
        }
        //
        taskItemIndexToTaskExecuteResultMap.putAll(executes(taskExecuteContext));
        // 汇总修改
        for (Integer taskItemIndex : taskItemIndexList) {
            if (taskItemIndex == null) {
                continue;
            }
            TaskItemExecuteResult taskExecuteResult = taskItemIndexToTaskExecuteResultMap.get(taskItemIndex);
            if (taskExecuteResult == null) {
                taskExecuteResult = new TaskItemExecuteResult(ReturnCode.SYSTEM_FAIL, "Can not find the corresponding SaturnJobReturn", TaskExecuteResultGroup.FAIL);
                taskItemIndexToTaskExecuteResultMap.put(taskItemIndex, taskExecuteResult);
            }
            updateTaskExecuteResult(taskExecuteContext, taskItemIndex, taskExecuteResult);
        }
        long end = System.currentTimeMillis();
        LogUtils.info(log, taskId, "{} finished, totalCost={}ms, return={}", taskId, (end - start), taskItemIndexToTaskExecuteResultMap);
    }

    protected void updateTaskExecuteResult(TaskExecuteContext taskExecuteContext, int taskItemIndex, TaskItemExecuteResult taskExecuteResult) {
        int successCount = 0;
        int errorCount = 0;
        if (ReturnCode.JOB_NO_COUNT != taskExecuteResult.getReturnCode()) {
            int errorGroup = taskExecuteResult.getErrorGroup();
            if (errorGroup == TaskExecuteResultGroup.SUCCESS) {
                successCount++;
            } else {
                if (errorGroup == TaskExecuteResultGroup.TIMEOUT) {
                    onTimeout(taskItemIndex);
                } else if (errorGroup == TaskExecuteResultGroup.FAIL_NEED_RAISE_ALARM) {
                    onNeedRaiseAlarm(taskItemIndex, taskExecuteResult.getReturnMsg());
                }
                errorCount++;
            }
        }
        // 为了展现分片处理失败的状态
        taskExecuteContext.getTaskItemIndexToExecuteResultMap().put(taskItemIndex, taskExecuteResult);
        // 执行次数加1
        ProcessCountStatistics.increaseTotalCountDelta(taskId, executorId);
        // 只要有出错和失败的分片，就认为是处理失败; 否则认为处理成功
        if (errorCount == 0 && successCount >= 0) {
            ProcessCountStatistics.incrementProcessSuccessCount(taskId, executorId, successCount);
        } else {
            ProcessCountStatistics.increaseErrorCountDelta(taskId, executorId);
            ProcessCountStatistics.incrementProcessFailureCount(taskId, executorId, errorCount);
        }
    }

    private void onNeedRaiseAlarm(int taskItemIndex, String returnMsg) {
    }

    private void onTimeout(int taskItemIndex) {
    }

    /**
     * 如果不存在该分片的running节点，又不是关闭了enabledReport的话，不继续执行；如果所有该executor分片running节点属于当前zk，继续执行；
     *
     * @param item 分片信息
     * @return 是否继续执行完complete节点，清空failover信息
     */
    private boolean checkIfZooKeeperLostAfterExecute(final Integer item) {
        CuratorFramework curatorFramework = executeService.getZookeeper().getCuratorFramework();
        try {
            String runningPath = TaskNodePaths.getTaskNodePath(taskId, ExecuteNode.getRunningNode(item));
            Stat stat = curatorFramework.checkExists().forPath(runningPath);
            long sessionId = curatorFramework.getZookeeperClient().getZooKeeper().getSessionId();
            // 有itemStat的情况
            if (stat != null) {
                long ephemeralOwner = stat.getEphemeralOwner();
                if (ephemeralOwner != sessionId) {
                    LogUtils.info(log, taskId, "item={} 's running node doesn't belong to current zk, node sessionid is {}, current zk sessionid is {}", item, ephemeralOwner, sessionId);
                    return false;
                } else {
                    return true;
                }
            }
            // 如果itemStat是空，要么是已经failover完了，要么是没有节点failover；两种情况都返回false
            LogUtils.info(log, taskId, "item={} 's running node is not exists, zk sessionid={} ", item, sessionId);
            return false;
        } catch (Throwable e) {
            LogUtils.error(log, taskId, e.getMessage(), e);
            return false;
        }
    }


    public void stop() {
        stopped = true;
    }


    public void forceStop() {
        forceStopped = true;
        synchronized (taskItemIndexToTaskItemExecuteTaskMap) {
            for (TaskItemExecuteTask taskItemFutureTask : taskItemIndexToTaskItemExecuteTaskMap.values()) {
                TaskItemExecute taskItemFutureTaskCaller = taskItemFutureTask.getTaskItemExecute();
                Thread currentThread = taskItemFutureTaskCaller.getCurrentThread();
                if (currentThread != null) {
                    try {
                        if (taskItemFutureTaskCaller.forceStop()) {
                            LogUtils.info(log, taskId, "Force stop job, jobName:{}, item:{}", taskId,
                                    taskItemFutureTaskCaller.getTaskItemIndex());
                            taskItemFutureTaskCaller.beforeForceStop();
                            TaskItemExecuteTask.killRunningBusinessThread(taskItemFutureTask);
                        }
                    } catch (Throwable t) {
                        LogUtils.error(log, taskId, t.getMessage(), t);
                    }
                }
            }
        }
    }


    public void abort() {
        aborted = true;
    }

    public TaskItemExecuteResult execute(final String taskId, TaskExecuteContext executeContext, final Integer taskItemIndex, final String taskItemValue, final TaskItemExecute taskItemFutureTaskCaller) throws Throwable {

        String taskClass = executeContext.getTaskConfiguration().getTaskClass();
        LogUtils.info(log, taskId, "Running SaturnJavaJob,  jobClass [{}], item [{}]", taskClass, taskItemIndex);
        try {
            BizTask bizTask = null;
            TaskItemExecuteResult taskExecuteResult = bizTask.execute(taskId, taskItemIndex, taskItemValue, taskItemFutureTaskCaller.getTaskItemExecuteContext());
            if (taskExecuteResult != null) {
                taskItemFutureTaskCaller.setHasTaskExecuteResult(true);
            }
            return taskExecuteResult;
        } catch (Exception e) {
            if (e.getCause() instanceof ThreadDeath) {
                throw e.getCause();
            }
            String message = logBusinessExceptionIfNecessary(taskId, e);
            return new TaskItemExecuteResult(ReturnCode.USER_FAIL, message, TaskExecuteResultGroup.FAIL);
        }
    }

    public void beforeTimeout(final String jobName, final Integer shardItem, final String shardParam,
                              TaskExecuteContext executeContext, final TaskItemExecute taskItemFutureTaskCaller) {
        BizTask bizTask = null;
        bizTask.beforeTimeout(jobName, shardItem, shardParam, taskItemFutureTaskCaller.getTaskItemExecuteContext());
    }

    public void postTimeout(final String jobName, final Integer shardItem, final String shardParam,
                            TaskExecuteContext executeContext, final TaskItemExecute taskItemFutureTaskCaller) {
        BizTask bizTask = null;
        bizTask.postTimeout(jobName, shardItem, shardParam, taskItemFutureTaskCaller.getTaskItemExecuteContext());
    }


    public void beforeForceStop(final String jobName, final Integer shardItem, final String shardParam,
                                TaskExecuteContext executeContext, final TaskItemExecute taskItemFutureTaskCaller) {
        BizTask bizTask = null;
        bizTask.beforeForceStop(jobName, shardItem, shardParam, taskItemFutureTaskCaller.getTaskItemExecuteContext());
    }

    public void postForceStop(final String jobName, final Integer shardItem, final String shardParam,
                              TaskExecuteContext executeContext, final TaskItemExecute taskItemFutureTaskCaller) {
        BizTask bizTask = null;
        bizTask.beforeForceStop(jobName, shardItem, shardParam, taskItemFutureTaskCaller.getTaskItemExecuteContext());
    }

    public void notifyTaskEnabled() {
        BizTask bizTask = null;
        String jobName = null;
        bizTask.onEnabled(jobName);
    }

    public void notifyTaskDisabled() {
        BizTask bizTask = null;
        String jobName = null;
        bizTask.onDisabled(jobName);
    }


    public Properties parseProperties(String path) {
        if (Strings.isNullOrEmpty(path)) {
            return null;
        }
        Properties kv = new Properties();
        String[] paths = path.split(",");
        if (paths != null && paths.length > 0) {
            for (String p : paths) {
                String[] tmps = p.split("=");
                if (tmps != null && tmps.length == 2) {
                    kv.put(tmps[0].trim(), tmps[1].trim());
                }
            }
        }

        return kv;
    }


    /**
     * 获取替换后的作业分片执行值
     *
     * @param taskParameter        作业参数
     * @param taskItemAndParameter 作业value
     * @return 替换后的值
     */
    protected String getRealTaskItemValue(String taskParameter, String taskItemAndParameter) {
        // 处理自定义参数
        Properties properties = parseProperties(taskParameter);
        int size = properties != null ? properties.size() : 0;
        final String string; // 作业分片的对应值
        if (size > 0) {
            // 有自定义参数, 解析完替换
            PropertyPlaceholderHelper propertyPlaceholderHelper = new PropertyPlaceholderHelper("{", "}");
            string = propertyPlaceholderHelper.replacePlaceholders(taskItemAndParameter, properties);
        } else {
            string = taskItemAndParameter;
        }
        return string.replaceAll("!!", "\"").replaceAll("@@", "=").replaceAll("##", ",");
    }

    public String logBusinessExceptionIfNecessary(String jobName, Throwable t) {
        String message = null;
        if (t instanceof ReflectiveOperationException) {
            Throwable cause = t.getCause();
            if (cause != null) {
                message = cause.toString();
            }
        }
        if (message == null) {
            message = t.toString();
        }
        LogUtils.error(log, jobName, message, t);
        return message;
    }

    /**
     * 实际处理逻辑
     *
     * @param executeContext 上下文
     * @return 每个分片返回一个SaturnJobReturn. 若为null，表示执行失败
     */
    protected Map<Integer, TaskItemExecuteResult> executes(final TaskExecuteContext executeContext) {
        final Map<Integer, TaskItemExecuteResult> taskItemIndexToTaskExecuteResultMap = new HashMap<Integer, TaskItemExecuteResult>();
        synchronized (taskItemIndexToTaskItemExecuteTaskMap) {
            taskItemIndexToTaskItemExecuteTaskMap.clear();
            //
            final String taskId = executeContext.getTaskId();
            // 处理自定义参数
            String taskParameter = executeContext.getTaskParameter();
            final int timeoutSeconds = taskConfigurationService.getTimeoutSeconds();
            // shardingItemParameters为参数表解析出来的Key/Value值
            Map<Integer, String> taskItemIndexToTaskItemAndParameterMap = executeContext.getTaskItemIndexToTaskItemAndParameterMap();
            for (final Map.Entry<Integer, String> entry : taskItemIndexToTaskItemAndParameterMap.entrySet()) {
                final Integer taskItemIndex = entry.getKey();
                String taskItemAndParameter = entry.getValue();
                try {
                    final String taskItemValue = getRealTaskItemValue(taskParameter, taskItemAndParameter); // 作业分片的对应值
                    TaskItemExecute taskItemExecute = new TaskItemExecute(taskId, this, taskItemIndex, taskItemValue, timeoutSeconds, executeContext);
                    TaskItemExecuteTask taskItemExecuteTask = new TaskItemExecuteTask(taskItemExecute, null);
                    Future<?> future = taskExecutor.getExecutorService().submit(taskItemExecuteTask);
                    if (timeoutSeconds > 0) {
                        TaskItemExecuteTimeoutTaskSchedulerExecutor.schedule(executeContext.getExecutorId(), taskItemExecuteTask, timeoutSeconds);
                    }
                    taskItemExecuteTask.setFuture(future);
                    taskItemIndexToTaskItemExecuteTaskMap.put(taskItemIndex, taskItemExecuteTask);
                } catch (Throwable t) {
                    LogUtils.error(log, taskId, t.getMessage(), t);
                    taskItemIndexToTaskExecuteResultMap.put(taskItemIndex, new TaskItemExecuteResult(ReturnCode.SYSTEM_FAIL, t.getMessage(), TaskExecuteResultGroup.FAIL));
                }
            }
        }

        for (Map.Entry<Integer, TaskItemExecuteTask> entry : taskItemIndexToTaskItemExecuteTaskMap.entrySet()) {
            Integer taskItemIndex = entry.getKey();
            TaskItemExecuteTask taskItemExecuteTask = entry.getValue();
            try {
                taskItemExecuteTask.getFuture().get();
            } catch (Exception e) {
                LogUtils.error(log, taskId, e.getMessage(), e);
                taskItemIndexToTaskExecuteResultMap.put(taskItemIndex, new TaskItemExecuteResult(ReturnCode.SYSTEM_FAIL, e.getMessage(), TaskExecuteResultGroup.FAIL));
                continue;
            }
            taskItemIndexToTaskExecuteResultMap.put(taskItemIndex, taskItemExecuteTask.getTaskItemExecute().getTaskExecuteResult());
        }
        synchronized (taskItemIndexToTaskItemExecuteTaskMap) {
            taskItemIndexToTaskItemExecuteTaskMap.clear();
        }
        return taskItemIndexToTaskExecuteResultMap;
    }


    public void resume() {
        stopped = false;
    }

    public void enableTask() {
        scheduler.getTrigger().enableTask();
    }

    public void disableTask() {
        scheduler.getTrigger().disableTask();
    }

    public void onResharding() {
        scheduler.getTrigger().onResharding();
    }

    public boolean isFailoverSupported() {
        return scheduler.getTrigger().isFailoverSupported();
    }


}
